package com.flink.cus.operators;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;

import org.apache.flink.api.common.ExecutionConfig.GlobalJobParameters;
import org.apache.flink.api.java.tuple.Tuple3;


public class Sink2 extends RichSinkFunction<Tuple3<Long, Float, Long>>{
    private StreamingRuntimeContext run_ctx;
	/**
	* open方法在sink第一次启动时调用，一般用于sink的初始化操作
	*/
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);      
        run_ctx = (StreamingRuntimeContext) getRuntimeContext();
        //System.out.print(run_ctx.getIndexOfThisSubtask()+",");
        GlobalJobParameters props = run_ctx.getExecutionConfig().getGlobalJobParameters();
        System.out.println("#sink::open()->"+props.toMap().get("srcTopic"));
    }

    /**
	* invoke方法是sink数据处理逻辑的方法，source端传来的数据都在invoke方法中进行处理
	* 第一个参数类型与RichSinkFunction<String>中的泛型对应。
    * 第二个参数为上下文信息
    */
    @Override
    public void invoke(Tuple3<Long, Float, Long> value, Context context) throws Exception {
    	System.out.println("#Max=>"+value+"@"+run_ctx.getIndexOfThisSubtask());        
    }

    /**
	* close方法在sink结束时调用，一般用于资源的回收操作
    */
    @Override
    public void close() throws Exception {        
        super.close();
    }
}
